package org.springframework.boot.autoconfigure.netty.channel;

import com.google.protobuf.Message;
import com.google.protobuf.MessageLite;
import io.netty.handler.codec.http.websocketx.BinaryWebSocketFrame;
import io.netty.handler.codec.http.websocketx.TextWebSocketFrame;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.boot.netty.codec.ProtoBufCodecBean;
import org.springframework.boot.netty.converter.MessageConverter;
import org.springframework.boot.netty.exception.ChannelInActiveException;

import java.net.InetSocketAddress;
import java.util.Collection;
import java.util.HashMap;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentMap;

/**
 * Author: huoxingzhi
 * Date: 2020/12/11
 * Email: hxz_798561819@163.com
 */
public class Channel {

    private io.netty.channel.Channel nettyChannel;

    public Collection<MessageConverter> messageConverters;

    private final static Logger logger = LoggerFactory.getLogger(Channel.class);


    public Channel(io.netty.channel.Channel nettyChannel, Collection<MessageConverter> messageConverters) {
        this.nettyChannel = nettyChannel;
        this.messageConverters = messageConverters;
    }

    public io.netty.channel.Channel getNettyChannel() {
        return this.nettyChannel;
    }

    public void sendToNettyWithWebSocket(Object object){
        /*InetSocketAddress socketAddress = (InetSocketAddress)nettyChannel.remoteAddress();
        String clientIp = socketAddress.getAddress().getHostAddress();
        sendToNettyWithWebSocket(object,clientIp);*/
        MessageConverter messageConverter = getMessageConverter(object.getClass(),String.class);

        this.nettyChannel.writeAndFlush(
                new TextWebSocketFrame(messageConverter.toMessage(String.class,object).toString()));

    }

    public void sendToNettyWithWebSocket(Object object,String receiverId){
        io.netty.channel.Channel receiverChannel = UserChannelRel.get(receiverId);
        if(receiverChannel==null){
            throw new ChannelInActiveException("用戶"+receiverId+"下线");
        }

        MessageConverter messageConverter = getMessageConverter(object.getClass(),String.class);

        receiverChannel.writeAndFlush(
                new TextWebSocketFrame(messageConverter.toMessage(String.class,object).toString()));

        //new BinaryWebSocketFrame()
    }

    /**
     * 任意java对象转为protobuf对象，传送给客户端
     * @param clazz
     * @param content
     * @param accessToken
     */
    public void sendToNettyWithProtoBuf(Class<? extends Message> clazz, Object content,String accessToken){
        InetSocketAddress socketAddress = (InetSocketAddress)nettyChannel.remoteAddress();
        String clientIp = socketAddress.getAddress().getHostAddress();
        sendToNettyWithProtoBuf(clazz,content,accessToken,clientIp);
    }

    public void sendToNettyWithProtoBuf(Class<? extends Message> clazz, Object content,String accessToken,String receiverId){
        io.netty.channel.Channel receiverChannel = UserChannelRel.get(receiverId);
        if(receiverChannel==null){
            throw new ChannelInActiveException("用戶"+receiverId+"下线");
        }
        MessageConverter messageConverter = getMessageConverter(content.getClass(), clazz);

        Message message = (Message) messageConverter.toMessage(clazz, content);

        ProtoBufCodecBean protoBufCodecBean = new ProtoBufCodecBean();
        Map<String,String> headers = new HashMap<>();
        headers.put("class",clazz.getSimpleName());
        headers.put("access_token",accessToken);
        protoBufCodecBean.setHeaders(headers);
        protoBufCodecBean.setMessage(message);
        receiverChannel.writeAndFlush(message);
    }

    private MessageConverter getMessageConverter(Class<?> source,Class<?> target){
        MessageConverter messageConverter = null;
        for (MessageConverter converter : messageConverters) {
            if (converter.toSupport(source,target)) {
                messageConverter = converter;
                break;
            }
        }
        return messageConverter;
    }

    /**
     * @Description: 用户id和channel的关联关系处理
     */
    public static class UserChannelRel {

        private static ConcurrentMap<String, io.netty.channel.Channel> manager = new ConcurrentHashMap<>(512);

        public static synchronized void put(String senderId, io.netty.channel.Channel channel) {
            if(manager.get(senderId)!=null && !manager.get(senderId).isActive()){
                manager.remove(senderId);
            }else {
                manager.put(senderId, channel);
            }

        }

        public static synchronized void remove(String senderId) {
            if(manager.get(senderId)!=null ){
                manager.remove(senderId);
            }
        }

        public static io.netty.channel.Channel get(String senderId) {
            return manager.get(senderId);
        }

        public static void output() {
            for (HashMap.Entry<String, io.netty.channel.Channel> entry : manager.entrySet()) {
                Channel.logger.info("UserId: {}, ChannelId: {}",entry.getKey(),entry.getValue().id().asLongText());
            }
        }
    }

}
